#define _GNU_SOURCE 

#include "syscore.h"
#include "apis.h"
#include "common.h"

#include <time.h>

#include <dirent.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>

#include <string.h>
#include <limits.h>
#include <stdio.h>
#include <stdlib.h>

#define SQLITE_HAS_CODEC 1
#include "sqlite3.h"
#include "../logServer/clogger.h"
#define DATAMLogDebug(fmt, args...) CLogger(MODULE_DATAM, __FILE__, __LINE__, LOG_DEBUG, fmt, ##args);
#define DATAMLogInfo(fmt, args...) CLogger(MODULE_DATAM, __FILE__, __LINE__, LOG_INFO, fmt, ##args);
#define DATAMLogNotice(fmt, args...) CLogger(MODULE_DATAM, __FILE__, __LINE__, LOG_NOTICE, fmt, ##args);
#define DATAMLogWarning(fmt, args...) CLogger(MODULE_DATAM, __FILE__, __LINE__, LOG_WARNING, fmt, ##args);
#define DATAMLogError(fmt, args...) CLogger(MODULE_DATAM, __FILE__, __LINE__, LOG_ERR, fmt, ##args);
#define DATAMLogCrit(fmt, args...) CLogger(MODULE_DATAM, __FILE__, __LINE__, LOG_CRIT, fmt, ##args);
#define DATAMLogAlert(fmt, args...) CLogger(MODULE_DATAM, __FILE__, __LINE__, LOG_ALERT, fmt, ##args);
#define DATAMLogEmerg(fmt, args...) CLogger(MODULE_DATAM, __FILE__, __LINE__, LOG_EMERG, fmt, ##args);
U32 maxStorHours = (180*24);
//#define MAX_ENTRY_PER_DAY (24)
//#define MAX_SECONDS (maxStorHours*3600)

#define nextHour(x) ((x)-(x)%3600 + 3600)

static char *restore_imei[15] = {"0", "00", "000", "0000", "00000", "000000", "0000000", "00000000", "000000000", "0000000000", "00000000000", "000000000000", "0000000000000", "00000000000000", "000000000000000"};
typedef struct tagDATAMSTATS {
    U32 numUeIDGotten;
    U32 numUeIDSaved;
    U32 numUeIDSend;
    U32 numBbuSaved[13];
}DATAMSTATS;

DATAMSTATS datamStats = {0};
//this PUHONEINFO is for GSM
#if 0
typedef struct tagPHONEINFO {
    U32 totallen;
    U64 uptime;
    char imsi[16];
    char imei[16];
    char mobilePhoneStandard[2];
    char tmsi[12];
} PHONEINFO;
#endif

//this PHONEINFO is for LTE
typedef struct tagPHONEINFO {
    U32 totallen;
    U64 uptime;
    int slotId;
    char imsi[16];
    char imei[16];
    char tmsi[12];
} PHONEINFO;

typedef struct tagRECODRinfo {
    U64 uptime;
    S32 slotId;
    char imsi[16];
    char imei[16];
} RECORDINFO;

RECORDINFO recordInfo;
#define MAX_RECORD_PER_MESSAGE 50

typedef struct tagPHONEINFOFLOWCTRL {
    BOOL started;
    U16 dstID;

    U64 begin_time;
    U64 end_time;
    U64 current_time;
    sqlite3* currentdb;
    char **dbResult;
    int nRow;
    int nColumn;
    int rowIdx;
    
    U32 maxnum;
    U32 widx;
    PHONEINFO phoneinfobuff[MAX_RECORD_PER_MESSAGE];
}PHONEINFOFLOWCTRL;

typedef struct tagDBDESC {
    U64 range_begin;
    U64 range_end;
} DBDESC;

#define DATASTOR_TIMER_RESOLUTION 5000
#define DATASTOR_HEATBEAT_INTERVAL 5000
#define FLOWCTRL_TIMEOUT 60000
PHONEINFOFLOWCTRL phoneInfoFlowCtrl;
void *FlowCtrlTimer = NULL;

extern char* data_storage_path;   
char recorddbpath[PATH_MAX] ;
DBDESC recorddb_desc;

sqlite3 *currentDB = NULL;
U32 currentDBHour = 0;
char* sql_creatdb = "PRAGMA synchronous = NORMAL;PRAGMA temp_store = MEMORY; CREATE TABLE IF NOT EXISTS PHONE_INFO( \
        TIMESTAMP INTEGER , \
        IMSI VARCHAR(16), \
        IMEI VARCHAR(16), \
        TMSI INTEGER,\
        SLOTID INTEGER);";

        
char* sql_creatidx= "CREATE INDEX IF NOT EXISTS TS_INDEX ON PHONE_INFO(TIMESTAMP);";  

char* sql_begin = "begin;";
char* sql_commit = "commit;";
char* sql_rollback = "rollback;";

char *dbstorekey = "123456";
extern PARADECODERMAP UEIDReportTable[4];

ERROR_CODE DATASTOR_decodeBeginTime(void *h, U8* paraValue, U32 paraLength );
ERROR_CODE DATASTOR_decodeEndTime(void *h, U8* paraValue, U32 paraLength );
ERROR_CODE DATASTOR_decodeNumPerMsg(void *h, U8* paraValue, U32 paraLength );
ERROR_CODE DATASTOR_decodeSteps(void *h, U8* paraValue, U32 paraLength );
ERROR_CODE DATASTOR_decodeHours(void *h, U8* paraValue, U32 paraLength );

void* pdatamHeartbeat = NULL;
void DATAM_heartbeat_timer(void* h)
{ 
    U32 rc = PARA_VALUE_SYSMONITOR_MODULE_OK;
    CM_SendMsg_1Para(MODULE_SYSMONITOR, MODULE_DATAM, MSG_TYPE_COMMAND, CMD_EVENT, CMD_CODE_SYSMONITOR_HEARTBEAT, 0,
        PARA_TYPE_SYSMONITOR_MODULE_STATE,sizeof(U32), (U8*)&rc);   
     
    MOD_timerRestart(pdatamHeartbeat, DATASTOR_HEATBEAT_INTERVAL);
}


typedef struct tagTRANSFERSTARTPARAS {
    U64 ts_begin;
    U64 ts_end;
    U32 numPerMsg;
    U16 dstID;
}TRANSFERSTARTPARAS;
TRANSFERSTARTPARAS flowctrlparaList = {0};


PARADECODERMAP transfercontroltable[] = {
    {PARA_TYPE_DATASTOR_BEGIN_TIME,DATASTOR_decodeBeginTime },
    {PARA_TYPE_DATASTOR_END_TIME, DATASTOR_decodeEndTime},
    {PARA_TYPE_DATASTOR_NUM_RECORDS_PER_MESSAGE, DATASTOR_decodeNumPerMsg}                      
};


typedef struct tagOBSOLETEPARAS {
    U32 steps;
}OBSOLETEPARAS;
OBSOLETEPARAS obsoleteparaList = {0};

PARADECODERMAP obsoletestepstable[] = {
    {PARA_TYPE_DATASTOR_OBSOLETE_STEPS, DATASTOR_decodeSteps }                     
};

typedef struct tagSTORRULEPARAS {
    U32 setmask;
    U32 hours;
}STORRULEPARAS;
STORRULEPARAS storruleparaList = {0};

PARADECODERMAP storruletable[] = {
    {PARA_TYPE_DATASTOR_STORAGE_HOURS, DATASTOR_decodeHours }                     
};



static BOOL needPostProc = FALSE;

RESULT DATASTOR_obsoleteData(void)
{
    
    struct tm thistm;
    char cmd[PATH_MAX + 32] ;
    time_t seconds = recorddb_desc.range_begin;

    if(recorddb_desc.range_begin/3600 >= recorddb_desc.range_end/3600)
        return RC_FAIL;
    
    if(recorddb_desc.range_begin/3600 == currentDBHour)
        return RC_FAIL;
    if(phoneInfoFlowCtrl.started)
        return RC_FAIL;
    {
        localtime_r(&seconds, &thistm);

        sprintf(cmd, "rm -rf %s/%04d%02d%02d/%02d", recorddbpath,1900+thistm.tm_year,1+thistm.tm_mon,thistm.tm_mday,thistm.tm_hour);
        system(cmd); 

        if(thistm.tm_hour == 23)
        {
            sprintf(cmd, "rm -rf %s/%04d%02d%02d", recorddbpath,1900+thistm.tm_year,1+thistm.tm_mon,thistm.tm_mday);
            system(cmd); 
        }

        recorddb_desc.range_begin = nextHour(seconds);        
        return RC_OK;
    }
    
}

void DATASTOR_updateTimeRange(U64 newtime)
{
    if(newtime > recorddb_desc.range_end)
    {
        recorddb_desc.range_end = newtime;
    }

    if((newtime < recorddb_desc.range_begin)
        ||(recorddb_desc.range_begin == 0) )
    {            
        recorddb_desc.range_begin = newtime;
    }

    while((recorddb_desc.range_begin/3600 + maxStorHours )< (recorddb_desc.range_end/3600) )
    {
        if(DATASTOR_obsoleteData() == RC_FAIL)
            break;
    }
}

ERROR_CODE DATASTOR_decodeHours(void *h, U8* paraValue, U32 paraLength )
{
    STORRULEPARAS * buff = (STORRULEPARAS*)h;

    if(paraLength == sizeof(U64))
    {
        buff->setmask |=0x1;
        buff->hours= *((U32*)paraValue);
        return PARA_VALUE_ERROR_NO_ERROR;
    }
    else
    {
        return PARA_VALUE_ERROR_INVALID_PARALENGTH;
    }
    
}

ERROR_CODE DATASTOR_decodeSteps(void *h, U8* paraValue, U32 paraLength )
{
    OBSOLETEPARAS * buff = (OBSOLETEPARAS*)h;

    if(paraLength == sizeof(U32))
    {
        U32 steps = *((U32*)paraValue);
        if( steps >0 )
        {    
            buff->steps= steps;
            return PARA_VALUE_ERROR_NO_ERROR;
        }
        else
        {
            return PARA_VALUE_ERROR_NUM_OUT_OF_RANGE;

        }
    }
    else
    {
        return PARA_VALUE_ERROR_INVALID_PARALENGTH;
    }
    
}

ERROR_CODE DATASTOR_decodeBeginTime(void *h, U8* paraValue, U32 paraLength )
{
    TRANSFERSTARTPARAS* buff = (TRANSFERSTARTPARAS*)h;

    if(paraLength == sizeof(U64))
    {
        buff->ts_begin = *((U64*)paraValue);
        return PARA_VALUE_ERROR_NO_ERROR;
    }
    else
    {
        return PARA_VALUE_ERROR_INVALID_PARALENGTH;
    }
    
}

ERROR_CODE DATASTOR_decodeEndTime(void *h, U8* paraValue, U32 paraLength )
{
    TRANSFERSTARTPARAS* buff = (TRANSFERSTARTPARAS*)h;

    if(paraLength == sizeof(U64))
    {
        buff->ts_end = *((U64*)paraValue);
        return PARA_VALUE_ERROR_NO_ERROR;
    }
    else
    {
        return PARA_VALUE_ERROR_INVALID_PARALENGTH;
    }

}

ERROR_CODE DATASTOR_decodeNumPerMsg(void *h, U8* paraValue, U32 paraLength )
{
    TRANSFERSTARTPARAS* buff = (TRANSFERSTARTPARAS*)h;

    if(paraLength == sizeof(U32))
    {
        U32 num = *((U32*)paraValue);
        if( (num >0) && (num <= MAX_RECORD_PER_MESSAGE) )
        {    
            buff->numPerMsg= num;
            return PARA_VALUE_ERROR_NO_ERROR;
        }
        else
        {
            return PARA_VALUE_ERROR_NUM_OUT_OF_RANGE;

        }
    }
    else
    {
        return PARA_VALUE_ERROR_INVALID_PARALENGTH;
    }
}


void DATASTOR_stopDataFlow()
{
    if(phoneInfoFlowCtrl.dbResult)
        sqlite3_free_table(phoneInfoFlowCtrl.dbResult);

    if(phoneInfoFlowCtrl.currentdb
        && (phoneInfoFlowCtrl.currentdb != currentDB))
    {   
        sqlite3_close(phoneInfoFlowCtrl.currentdb);
    }
    
    phoneInfoFlowCtrl.dstID = 0;

    phoneInfoFlowCtrl.dbResult = NULL;
    phoneInfoFlowCtrl.currentdb = NULL;
    phoneInfoFlowCtrl.started = FALSE;
    
    phoneInfoFlowCtrl.widx = 0;
    phoneInfoFlowCtrl.rowIdx = 0;
    


}

void DATASTOR_FlowCtrlTimeout(void *h)
{
    if(phoneInfoFlowCtrl.started)
    {
        DATASTOR_stopDataFlow();
    }
}

RESULT DATASTOR_initFlowCtrl()
{
    phoneInfoFlowCtrl.started = FALSE;
    phoneInfoFlowCtrl.dstID = 0;
    phoneInfoFlowCtrl.widx = 0;
    phoneInfoFlowCtrl.rowIdx = 0;
    phoneInfoFlowCtrl.nRow= 0;
    phoneInfoFlowCtrl.nColumn = 0;
    phoneInfoFlowCtrl.dbResult = NULL;
    phoneInfoFlowCtrl.currentdb = NULL;
                   
    return RC_OK;
}

RESULT DATASTOR_buildOneRecord(void )
{
    int len = 0;
    PHONEINFO *record = &phoneInfoFlowCtrl.phoneinfobuff[phoneInfoFlowCtrl.widx++];
    U32 start_idx = phoneInfoFlowCtrl.nColumn * (++phoneInfoFlowCtrl.rowIdx);
    char **dbResult =  &phoneInfoFlowCtrl.dbResult[start_idx];
    char nchar[1]={'\0'};    

    record->uptime = strtoull(dbResult[0],NULL,10);
    record->slotId = atoi(dbResult[4]);

 #if 1 
    len = 15 - strlen(dbResult[1]);
    if (len > 0)
    {
        memcpy(record->imsi, restore_imei[len - 1], len);
        sprintf(record->imsi + len, "%s", dbResult[1]);
    }
    else
        memcpy(record->imsi, dbResult[1], 15);
#endif
    //strcpy(record->imsi,(dbResult[1]==NULL?nchar:dbResult[1])); 
    strcpy(record->imei,(dbResult[2]==NULL?nchar:dbResult[2]));
    strcpy(record->tmsi,(dbResult[3]==NULL?nchar:dbResult[3]));
 
    record->totallen = sizeof(U64) + sizeof(int) + strlen(record->imsi)+strlen(record->imei)+strlen(record->tmsi) + 3;
    return RC_OK;
}


RESULT DATASTOR_openHourDB(PHONEINFOFLOWCTRL* fc)
{
    struct tm thistm ; 
    char filename[PATH_MAX];
    int rc = 0;

    if((currentDBHour == (fc->current_time/3600))
        && currentDB)
    {
           fc->currentdb   =   currentDB;                       
    }
    else
    {
        localtime_r((time_t*)(&fc->current_time),&thistm);
        
        sprintf(filename, "%s/%04d%02d%02d/%02d", recorddbpath,1900+thistm.tm_year,1+thistm.tm_mon,thistm.tm_mday,thistm.tm_hour);
        //printf("open db : %s \n",filename);
        rc =sqlite3_open(filename,&fc->currentdb );
        if(rc )
        {
            DATAMLogError("Can't open database: %s", sqlite3_errmsg(fc->currentdb));
            sqlite3_close(fc->currentdb);
            fc->currentdb = NULL;
            
            return RC_FAIL;
        }

       // rc = sqlite3_key(fc->currentdb,dbstorekey,6);
        if(SQLITE_OK != rc)
        {
            DATAMLogError("Can't open database: %s", sqlite3_errmsg(fc->currentdb));
            sqlite3_close(fc->currentdb);
            fc->currentdb = NULL;
            
            return RC_FAIL;
        }
        
    }
    return RC_OK;
}


RESULT DATASTOR_getRecordWithinOneSecond(PHONEINFOFLOWCTRL* fc)
{
    int rc = 0;
    char sql[256];

    sprintf(sql,"SELECT * FROM PHONE_INFO WHERE TIMESTAMP = %llu;",fc->current_time);
    //printf("sql : %s \n",sql);

    rc = sqlite3_get_table(fc->currentdb,sql,&fc->dbResult,&fc->nRow,&fc->nColumn,NULL);
    if(rc)
    {
        //dbgprintf(DEBUG_ERROR,stderr, "Get phone info failed: %p %p %s : %s\n",fc->currentdb,currentDB,sqlite3_errmsg(fc->currentdb), sql);
        sqlite3_free_table(fc->dbResult);
        fc->dbResult = NULL;
        return RC_FAIL;
    }
    
    return RC_OK;
    
}


void DATASTOR_sendDataIndication(BOOL bEOF)
{
    U64 msgbuff[MSG_BUFF_SIZE/8]={0};
    PMSGHEADER pMsgHeader =(PMSGHEADER) msgbuff;
    PCMDHEADER pCmdHeader = (PCMDHEADER)( ((U8*)msgbuff) + sizeof(MSGHEADER) );
    PPARAM pParam = (PPARAM)(pCmdHeader + 1);
    U8 *paraValue = NULL;
    int i = 0;
    
    pMsgHeader->dstID = phoneInfoFlowCtrl.dstID;
    pMsgHeader->srcID = MODULE_DATAM;
    pMsgHeader->msgType = MSG_TYPE_COMMAND;

    pCmdHeader->cmdCode = CMD_CODE_PHONE_INFO_DATA;
    pCmdHeader->cmdIndex = gCmdIndex++;
    pCmdHeader->cmdType = CMD_EVENT;
    pCmdHeader->cmdLength = 0; 


    pCmdHeader->cmdLength += CM_paraPackPara(&pParam,PARA_TYPE_DATASTOR_HAS_MORE_DATA,sizeof(U32),&paraValue);
    *((U32*)paraValue) = !bEOF;
        
    for(i=0;i<phoneInfoFlowCtrl.widx;i++)
    {
        pCmdHeader->cmdLength += CM_paraPackPara(&pParam,PARA_TYPE_DATASTOR_PHONE_INFO,
            phoneInfoFlowCtrl.phoneinfobuff[i].totallen,
            &paraValue);
        memcpy(paraValue,&phoneInfoFlowCtrl.phoneinfobuff[i].uptime,phoneInfoFlowCtrl.phoneinfobuff[i].totallen );

    }
    phoneInfoFlowCtrl.widx = 0;
    
    pMsgHeader->msgLength = sizeof(MSGHEADER) + sizeof(CMDHEADER)+ pCmdHeader->cmdLength;
    CM_sendMsg(pMsgHeader,pCmdHeader);



}

void DATASTOR_dataFlowLoop()
{
    BOOL bEOF;
    while(phoneInfoFlowCtrl.current_time < phoneInfoFlowCtrl.end_time)
    {
        int loopcount = 0;
        int i=0;
        RESULT rc = RC_OK;    
        
        if(phoneInfoFlowCtrl.currentdb == NULL)
        {
            if(RC_OK != DATASTOR_openHourDB(&phoneInfoFlowCtrl))
            {
                //move to next hour
                phoneInfoFlowCtrl.current_time = nextHour(phoneInfoFlowCtrl.current_time);
                continue;
            }            
        }
        
        if(phoneInfoFlowCtrl.dbResult == NULL)
        {
            rc = DATASTOR_getRecordWithinOneSecond(&phoneInfoFlowCtrl);
            
        }
        
        if((RC_OK == rc) && (phoneInfoFlowCtrl.nRow > 0))
        {
            #if 0
            int i, j;
            printf("row: %d, col:%d \n",phoneInfoFlowCtrl.nRow,phoneInfoFlowCtrl.nColumn);
                
            for(i = 0; i < phoneInfoFlowCtrl.nRow; i++)
            {
                for(j = 0; j < phoneInfoFlowCtrl.nColumn; j++)
                {
                
                    printf("%s ",phoneInfoFlowCtrl.dbResult[i*phoneInfoFlowCtrl.nColumn+j]);
                
                }
                printf("\n");

            }
            #endif
            if(((phoneInfoFlowCtrl.nRow - phoneInfoFlowCtrl.rowIdx) + phoneInfoFlowCtrl.widx) <= phoneInfoFlowCtrl.maxnum)
            {
                loopcount =  phoneInfoFlowCtrl.nRow - phoneInfoFlowCtrl.rowIdx;
            } 
            else
            {
                loopcount =  phoneInfoFlowCtrl.maxnum - phoneInfoFlowCtrl.widx ;
            }

            
            for(i =0 ;i<loopcount; i++ )
            {
                DATASTOR_buildOneRecord();
            }

            if(phoneInfoFlowCtrl.rowIdx == (phoneInfoFlowCtrl.nRow))
            {
                sqlite3_free_table(phoneInfoFlowCtrl.dbResult);
                phoneInfoFlowCtrl.dbResult = NULL;
                phoneInfoFlowCtrl.rowIdx = 0;
                phoneInfoFlowCtrl.current_time++;
                
                if(phoneInfoFlowCtrl.current_time % 3600 == 0)
                {
                       if(phoneInfoFlowCtrl.currentdb
                            && (phoneInfoFlowCtrl.currentdb != currentDB))
                       {
                            sqlite3_close(phoneInfoFlowCtrl.currentdb);
                       }
                       
                       phoneInfoFlowCtrl.currentdb = NULL;
                }
                
                 
            }

            if(phoneInfoFlowCtrl.widx == phoneInfoFlowCtrl.maxnum)
            {                
                break;
            }

        }
        else
        {
            phoneInfoFlowCtrl.current_time++;

            //move to next seconds 
            
            if(phoneInfoFlowCtrl.dbResult)
            {
                sqlite3_free_table(phoneInfoFlowCtrl.dbResult);
                phoneInfoFlowCtrl.dbResult = NULL;
                phoneInfoFlowCtrl.rowIdx = 0;
            }
            
             
            if((phoneInfoFlowCtrl.current_time%3600) == 0)
            {
                if(phoneInfoFlowCtrl.currentdb
                            && (phoneInfoFlowCtrl.currentdb != currentDB))
                {
                    sqlite3_close(phoneInfoFlowCtrl.currentdb);
                }
                       
                phoneInfoFlowCtrl.currentdb = NULL;
            }
        }

        
        
    }
    bEOF = (phoneInfoFlowCtrl.current_time >= phoneInfoFlowCtrl.end_time);
    DATASTOR_sendDataIndication(bEOF);
    
    if(bEOF)
    {           
        if(phoneInfoFlowCtrl.dbResult)
            sqlite3_free_table(phoneInfoFlowCtrl.dbResult);

        if(phoneInfoFlowCtrl.currentdb
            && (phoneInfoFlowCtrl.currentdb != currentDB))
        {
            sqlite3_close(phoneInfoFlowCtrl.currentdb);
        }

        phoneInfoFlowCtrl.dbResult = NULL;
        phoneInfoFlowCtrl.currentdb = NULL;
        phoneInfoFlowCtrl.started = FALSE;
                
    }
    else
    {
        FlowCtrlTimer = MOD_timerStart(MODULE_DATAM,FLOWCTRL_TIMEOUT,DATASTOR_FlowCtrlTimeout,NULL);
    
    }
}


void DATASTOR_startDataFlow(TRANSFERSTARTPARAS *para)
{        
	
    //dbgprintf(DEBUG_INFO, stdout, "---%llu  %llu  %llu  %llu ----\n",recorddb_desc.range_begin,recorddb_desc.range_end,para->ts_begin,para->ts_end);
    phoneInfoFlowCtrl.dstID = para->dstID;
    if((para->ts_end <= recorddb_desc.range_begin)
       ||(para->ts_begin> recorddb_desc.range_end)
        || (recorddb_desc.range_begin== 0)
       || (recorddb_desc.range_end == 0))  // no record
    {
        phoneInfoFlowCtrl.widx = 0 ;
        DATASTOR_sendDataIndication(TRUE);
        phoneInfoFlowCtrl.dstID = 0;
        return; 
    }

    phoneInfoFlowCtrl.started = TRUE;

    if(para->ts_begin < recorddb_desc.range_begin)
    {
        phoneInfoFlowCtrl.begin_time = recorddb_desc.range_begin;
    }
    else
    {
        phoneInfoFlowCtrl.begin_time = para->ts_begin;
    }
    
    if(para->ts_end > recorddb_desc.range_end)
    {
        phoneInfoFlowCtrl.end_time = recorddb_desc.range_end + 1;
    }
    else
    {
        phoneInfoFlowCtrl.end_time = para->ts_end;
    }
#if 0
    dbgprintf(DEBUG_INFO, stdout, "rang start  %d  rang  end %d pl start %d pl  end %d \n", para->ts_begin, para->ts_end, phoneInfoFlowCtrl.begin_time,  phoneInfoFlowCtrl.end_time);
#endif
    phoneInfoFlowCtrl.maxnum = para->numPerMsg;
    phoneInfoFlowCtrl.current_time = phoneInfoFlowCtrl.begin_time;
#if 0
    dbgprintf(DEBUG_INFO, stdout, "start begin %llu -> %llu \n", phoneInfoFlowCtrl.current_time , phoneInfoFlowCtrl.end_time );
#endif
    DATASTOR_dataFlowLoop();

}

RESULT DATASTOR_savePhoneInfo(U64 uptime,char *imsirecord, char *imeirecord,char *tmsirecord, U8 slotId)
{
    struct tm thistm;
//    time_t thistime;
//    char subpathname[12];
    char subpath[PATH_MAX];
//    char file[12];
    char filename[PATH_MAX]; 
    char sql[512];
    DIR *pDir;
    U32 thishour;
    sqlite3 *db=NULL;
    int rc = 0;

    if(uptime < 3600)  
        return RC_FAIL;

    thishour = uptime/3600;
#if 0
    dbgprintf(DEBUG_INFO, stdout, "thishourt ====== %d uptime == %d  currentDBHours  = %d\n", thishour, uptime, currentDBHour);
    char buf111[128];
    sprintf(buf111, "%d", currentDBHour);
    dbgprintf(DEBUG_INFO, stdout, "dbhour === %s", buf111);
#endif
    if((currentDBHour ==0) || (currentDBHour != thishour))
    {// first time
 //   dbgprintf(DEBUG_INFO, stdout, "thishourt111111 ====== %d uptime == %d \n", thishour, uptime);
        if(currentDB != NULL)
        {
            if( (phoneInfoFlowCtrl.started)
                && (phoneInfoFlowCtrl.currentdb == currentDB) )
            {
                    //used by flow control, do not close.
            }
            else
            {
                rc = sqlite3_exec(currentDB, sql_commit, 0, 0, NULL);
                if(rc)
                {                        
                    rc = sqlite3_exec(currentDB, sql_rollback, 0, 0, NULL);
                    DATAMLogError("sqlite transaction commint failed %x ",rc);
                    sqlite3_close(currentDB);
                    return RC_FAIL;;
                }
                sqlite3_close(currentDB);
            }
            //reset to initial value
            currentDB = NULL;
            currentDBHour = 0;
        }
        localtime_r((time_t*)(&uptime), &thistm);
        
        sprintf(subpath, "%s/%04d%02d%02d", recorddbpath,1900+thistm.tm_year,1+thistm.tm_mon,thistm.tm_mday);
        sprintf(filename, "%s/%02d",subpath,thistm.tm_hour);
     
        #if 0
        pDir = opendir(recorddbpath);
        if(pDir == NULL)
        {
           if (mkdir(recorddbpath,0777) < 0)
           {
               dbgprintf(DEBUG_ERROR,stderr,"creat dir failed!!!\n",NULL);
               return RC_FAIL;
           }
    
        }
        else
        {
            closedir(pDir);
        }
        #endif
        pDir = opendir(subpath);
        if(pDir == NULL)
        {
           if (mkdir(subpath,0777) != 0)
           {
               DATAMLogError("creat dir %s failed!!!",subpath);
               return RC_FAIL;
           }

        }
        else
        {
            closedir(pDir);

        }

        rc =sqlite3_open(filename,&db );
        if(rc )
        {
            DATAMLogError("Can't open database: %s", sqlite3_errmsg(db));
            sqlite3_close(db);
            return RC_FAIL;
        }
        
      //  rc = sqlite3_key(db,dbstorekey,6);
        if(SQLITE_OK != rc)
        {
            DATAMLogError("Can't open database: %s", sqlite3_errmsg(db));
            sqlite3_close(db);            
            return RC_FAIL;
        }

        rc = sqlite3_exec(db,sql_creatdb,0,0,NULL);
        if(rc)
        {
            DATAMLogError("init0 result %x ",rc);
            sqlite3_close(db);
            return RC_FAIL;
        }

        rc = sqlite3_exec(db,sql_creatidx,0,0,NULL);
        if(rc)
        {            
            DATAMLogError("init1 result %x",rc);
            sqlite3_close(db);
            return RC_FAIL;
        }

        currentDBHour = thishour;
 
        currentDB = db;
       if ( currentDB != NULL  )
       {
           rc = sqlite3_exec(currentDB,sql_begin,0,0,NULL);
            if(rc)
            {            
                DATAMLogError("sqlite transaction failed %x ",rc);
                sqlite3_close(currentDB);
                return RC_FAIL;
            }
       }
    }

    if(currentDB == NULL)
    {
        DATAMLogError("Unexpected error in opendb", NULL);
        return RC_FAIL;
    }
    else
    {       

           sprintf(sql,"INSERT INTO PHONE_INFO VALUES(%llu,'%s','%s',%s,%d);",

            uptime,
            (imsirecord==NULL?"NULL":imsirecord), 
            (strlen(imeirecord)<=0?"NULL":imeirecord),
            (tmsirecord==NULL?"NULL":tmsirecord),
            (slotId));
            DATAMLogDebug("slot[%d] imsi:%s ,imei:%s ", slotId, imsirecord, imeirecord);
        rc = sqlite3_exec(currentDB,sql,0,0,NULL);
        if(rc)
        {
            DATAMLogError("Insert phone info failed %x", rc);
            return RC_FAIL;
        }
        datamStats.numUeIDSaved++;
#ifdef CDMA_DEVICE
        slotId = 10;
#endif 
        datamStats.numBbuSaved[slotId - 1]++;
        DATASTOR_updateTimeRange(uptime);
    }
    
    return RC_OK;
}



void DATASTOR_setProc(PCMDHEADER pCmd, PCMDHEADER pAck)
{
    switch(pCmd->cmdCode)
    {
        case CMD_CODE_PHONE_INFO_TRANSFER_START:
            {
                PMSGHEADER pMsgHdr = (PMSGHEADER)((U8*)pCmd - sizeof(MSGHEADER));
                memset(&flowctrlparaList,0,sizeof(TRANSFERSTARTPARAS));
                if(RC_OK == CM_paraListDecoder(pCmd,pAck, transfercontroltable,sizeof(transfercontroltable)/sizeof(PARADECODERMAP),&flowctrlparaList,FALSE)) 
                {
                    if(flowctrlparaList.ts_begin == 0) // this parameter is must 
                    {
                        flowctrlparaList.ts_begin = recorddb_desc.range_begin;                        
                    }

                    if(flowctrlparaList.ts_end == 0)
                    {
                        flowctrlparaList.ts_end = recorddb_desc.range_end + 1;
                    }
                    
                    if(flowctrlparaList.ts_begin >= flowctrlparaList.ts_end)
                    {
                        
                        pAck->cmdLength = CM_paraPackError((PPARAM)(pAck+1), PARA_VALUE_ERROR_END_NOT_GREATER_THAN_BEGIN);
                        DATAMLogError("Error in cmdType=%02x cmdCode=%04x. Error=%x",pCmd->cmdType,pCmd->cmdCode,PARA_VALUE_ERROR_END_NOT_GREATER_THAN_BEGIN);
                        return;
                    }

                    
                    if(flowctrlparaList.numPerMsg== 0)
                    {
                        flowctrlparaList.numPerMsg = 50;
                    }

                    if(phoneInfoFlowCtrl.started)
                    {
                        pAck->cmdLength = CM_paraPackError((PPARAM)(pAck+1), PARA_VALUE_ERROR_DATAFLOW_BUSY);
                        DATAMLogError("Error in cmdType=%02x cmdCode=%04x. Error=%x",pCmd->cmdType,pCmd->cmdCode,PARA_VALUE_ERROR_DATAFLOW_BUSY);
                        return;
                    }

                    flowctrlparaList.dstID = pMsgHdr->srcID;
                    needPostProc = TRUE;
                    pAck->cmdLength = CM_paraPackU32((PPARAM)(pAck+1), PARA_TYPE_RESULT,PARA_VALUE_RESULT_SUCCESS);   

                }
            }
            break;
        case CMD_CODE_PHONE_INFO_TRANSFER_STOP:
            {
                PMSGHEADER pMsgHdr = (PMSGHEADER)((U8*)pCmd - sizeof(MSGHEADER));
                if(phoneInfoFlowCtrl.started)
                {
                    if(pMsgHdr->srcID == phoneInfoFlowCtrl.dstID)
                    {
                        MOD_timerClear(MODULE_DATAM,FlowCtrlTimer);
                        FlowCtrlTimer = NULL;
                        DATASTOR_stopDataFlow();
                        
                        pAck->cmdLength = CM_paraPackU32((PPARAM)(pAck+1), PARA_TYPE_RESULT,PARA_VALUE_RESULT_SUCCESS); 
                    }
                    else
                    {
                        pAck->cmdLength = CM_paraPackError((PPARAM)(pAck+1), PARA_VALUE_ERROR_NOT_DATAFLOW_PEER);
                    }
                }
                else
                {
                    pAck->cmdLength = CM_paraPackU32((PPARAM)(pAck+1), PARA_TYPE_RESULT,PARA_VALUE_RESULT_SUCCESS); 
                }
            }

            break;
        case CMD_CODE_OBSOLETE_PHONE_INFO:
            obsoleteparaList.steps = 0;
            if(RC_OK == CM_paraListDecoder(pCmd,pAck, obsoletestepstable,sizeof(obsoletestepstable)/sizeof(PARADECODERMAP),&obsoleteparaList,FALSE)) 
            {
                U32 steps= obsoleteparaList.steps;
                while(TRUE) 
                {
                    if(DATASTOR_obsoleteData() == RC_FAIL) 
                        break;
                }

                if(recorddb_desc.range_begin/3600 < recorddb_desc.range_end/3600 && recorddb_desc.range_begin/3600 != currentDBHour)
                {
                    pAck->cmdLength = CM_paraPackError((PPARAM)(pAck+1), PARA_VALUE_ERROR_COMPLETED_PARTLY);
                    DATAMLogNotice("clean data failed");
                }
                else
                {
                    pAck->cmdLength = CM_paraPackU32((PPARAM)(pAck+1), PARA_TYPE_RESULT,PARA_VALUE_RESULT_SUCCESS); 
                    DATAMLogNotice("clean data success");
                }
            }
            break;
        case CMD_CODE_PHONE_INFO_STORAGE_HOURS:
            storruleparaList = (STORRULEPARAS){0,0};
            if(RC_OK == CM_paraListDecoder(pCmd,pAck, storruletable,sizeof(storruletable)/sizeof(PARADECODERMAP),&storruleparaList,FALSE)) 
            {
                
                if(storruleparaList.setmask & 0x1)
                {    
                    maxStorHours = storruleparaList.hours;

                    while((recorddb_desc.range_begin/3600 + maxStorHours )< (recorddb_desc.range_end/3600) )
                    {
                        if(DATASTOR_obsoleteData() == RC_FAIL)
                            break;
                    }
                        
                    pAck->cmdLength = CM_paraPackU32((PPARAM)(pAck+1), PARA_TYPE_RESULT,PARA_VALUE_RESULT_SUCCESS); 
                }
                else
                { 
                    pAck->cmdLength = CM_paraPackError((PPARAM)(pAck+1), PARA_VALUE_ERROR_NOT_ENOUGH_PARA);
                }
            }
            break;
#ifdef MODULE_TEST
        case CMD_CODE_SYSMONITOR_TEST_MODULE:
            DATAMLogError("---------- test module datam ---------------- %04x", CMD_CODE_SYSMONITOR_TEST_MODULE);
            while(1)
            {

            }
            break;
#endif            
        default:
            pAck->cmdLength = CM_paraPackError((PPARAM)(pAck+1), PARA_VALUE_ERROR_UNKNOWN_CMDCODE);
            DATAMLogError("Unknown SET cmd cmdCode=%04x",pCmd->cmdCode);
            break;
    }

}

void DATASTOR_getProc(PCMDHEADER pCmd, PCMDHEADER pAck)
{
    switch(pCmd->cmdCode)
    {
        
        default:
            pAck->cmdLength = CM_paraPackError((PPARAM)(pAck+1), PARA_VALUE_ERROR_UNKNOWN_CMDCODE);            
            DATAMLogError("Unknown GET cmd cmdCode=%04x",pCmd->cmdCode);
            break;        
    }

}


//for indication, error do nothing
void DATASTOR_ueIDReport(PCMDHEADER pCmd)
{
/*
    UEIDPARALIST paraList = {0};
   
    if(RC_OK == CM_paraListDecoder(pCmd,NULL, UEIDReportTable,sizeof(UEIDReportTable)/sizeof(PARADECODERMAP),&paraList,TRUE))
    {
        if(paraList.uptime == 0)
        {
            dbgprintf(DEBUG_ERROR,stderr,"Error in cmdType=%02x cmdCode=%04x. Error=%x\n",pCmd->cmdType,pCmd->cmdCode,PARA_VALUE_ERROR_NOT_ENOUGH_PARA);
            return;
        }
        datamStats.numUeIDGotten++;
        dbgprintf(DEBUG_INFO, stdout, "datastore imsi ===  %s imei === %s slotid == %d\n", paraList.imsirecord, paraList.imeirecord, paraList.slotId);
//use for GSM PALTFORM
#if 0
        dbgprintf(DEBUG_INFO, stdout, "standard ==== %s\n", paraList.Standard);
        DATASTOR_savePhoneInfo(paraList.uptime,paraList.imsirecord, paraList.imeirecord,NULL, paraList.Standard);
#endif
        DATASTOR_savePhoneInfo(paraList.uptime,paraList.imsirecord, paraList.imeirecord,NULL, paraList.slotId);       

    }*/

    PPARAM pParam = (PPARAM) (pCmd + 1);
    U32 len = pCmd->cmdLength;
    U32 paraType;
    U32 paraLength;
    U8* paraValue;
    UNPACK_RESULT uprc;
    S32 rc = 0;
    
    if ( currentDB != NULL  )
    {
        rc = sqlite3_exec(currentDB,sql_begin,0,0,NULL);
        if(rc)
        {            
            DATAMLogError("sqlite transaction failed %x ",rc);
            sqlite3_close(currentDB);
            return ;
        }
    }

    while( (uprc=CM_paraUnpackPara(&pParam, &len, &paraType, &paraLength, &paraValue)) == UNPACK_OK )
    {
        switch(paraType)
        {
            case PARA_TYPE_DATAPROC_PHONE_INFO:
                memset(&recordInfo, 0, sizeof(RECORDINFO));
                recordInfo.uptime = *(U64 *)paraValue;
                paraValue += sizeof(U64);
                recordInfo.slotId = *(U16 *)paraValue;
                paraValue += sizeof(S32);
                strcpy((char *)recordInfo.imsi, (char *)paraValue);
                paraValue += (strlen((char *)paraValue) + 1);
                strcpy((char *)recordInfo.imei, (char *)paraValue);
                DATASTOR_savePhoneInfo(recordInfo.uptime, recordInfo.imsi, recordInfo.imei,NULL, recordInfo.slotId); 
                break;

            default: 
                DATAMLogError("unknow paraType %0x in unpack UeId ", paraType);
                break; 

        }

    }

    if ( currentDB != NULL  )
    {
        rc = sqlite3_exec(currentDB, sql_commit, 0, 0, NULL);
        if(rc)
        {                        
            rc = sqlite3_exec(currentDB, sql_rollback, 0, 0, NULL);
            DATAMLogError("sqlite transaction commint failed %x ",rc);
            sqlite3_close(currentDB);
            return ;
        }
    }
   
    if ( uprc == UNPACK_ERROR )
    {
        DATAMLogError(" recv unpack msg ");
        return;
    }

    
}

void DATASTOR_eventProc(PCMDHEADER pCmd)
{
    switch(pCmd->cmdCode)
    {
        //case CMD_GSM_EVT_PHONE:
        case CMD_CODE_DATAPRO_PHONE_INFO:
            
            DATASTOR_ueIDReport(pCmd);
            break;
        default:
            DATAMLogNotice("Unknown EVENT cmdCode=%04x",pCmd->cmdCode);            
            break;
        

    }

}

void DATASTOR_eventAckProc(PCMDHEADER pCmd)
{
    

    switch(pCmd->cmdCode)
    {
        case CMD_CODE_PHONE_INFO_DATA:
            {
                PMSGHEADER pMsgHdr = (PMSGHEADER)((U8*)pCmd - sizeof(MSGHEADER));
                if(phoneInfoFlowCtrl.started && (phoneInfoFlowCtrl.dstID == pMsgHdr->srcID))
                {
                    MOD_timerClear(MODULE_DATAM,FlowCtrlTimer);
                    FlowCtrlTimer = NULL;
                    
                    DATASTOR_dataFlowLoop();
                }
            }
            break;
        default:
            DATAMLogNotice("Unknown EVENT cmdCode=%04x",pCmd->cmdCode);            
            break;
        

    }

}

void DATASTOR_postProc(void)
{

    if(needPostProc)
    {        
        needPostProc = FALSE;
        DATASTOR_startDataFlow(&flowctrlparaList);
    }
        
}


void* DATASTOR_main(void*h)
{
	CMMODINFO cmModInfo = {0};
    pthread_setname_np(pthread_self(), "datam");
    pdatamHeartbeat = MOD_timerStart(MODULE_DATAM, DATASTOR_HEATBEAT_INTERVAL, DATAM_heartbeat_timer, NULL);
    cmModInfo.ModID = MODULE_DATAM;
    cmModInfo.useTimer = TRUE;
    cmModInfo.setProc = DATASTOR_setProc;
    cmModInfo.getProc = DATASTOR_getProc;
    cmModInfo.eventProc = DATASTOR_eventProc;    
    cmModInfo.eventAckProc = DATASTOR_eventAckProc;    
    cmModInfo.postProc = DATASTOR_postProc;
    
    CM_cmdDispatcher(&cmModInfo);

    
    return NULL;

}


RESULT DATASTOR_validateDirName(char *pFname)
{
    //20150908
    struct tm tm = {0};
    time_t thistime;
    
    if(strlen(pFname) != 8)
        return RC_FAIL;

    if(strptime(pFname,"%Y%m%d",&tm) == NULL)
        return RC_FAIL;

    thistime = mktime(&tm);
    if(thistime < 0 )
        return RC_FAIL;
#if 0
    if((thistime/3600/24 + MAX_DAYS) < baseTime/3600/24)
    {// too long ago
        return RC_FAIL;
    }
#endif    
    return RC_OK;
}

RESULT DATASTOR_validateFileName(char* pPathName, char *pFname,time_t* filetime)
{
    //00 ~ 23
    int i=0;
    char buf[20];
    struct tm tm = {0};
    time_t thistime;
    int fileLen = strlen(pFname);

    if(fileLen < 2)
        return RC_FAIL;

    if(fileLen != 2)
    {
        if(strstr(pFname, "-journal") != &pFname[2])
        {
            return RC_FAIL;
        }        
    }
    
    if(pFname[0] != '0' && pFname[0] != '1' && pFname[0] != '2')
    {
        return RC_FAIL;
    }

    if((pFname[1] < '0') || (pFname[1] > '9'))
    {
        return RC_FAIL;
    }

    i= atoi(pFname);
    if((i<0) ||(i >= 24))
    {
        return RC_FAIL;
    }
    
    sprintf(buf,"%s%02d",pPathName,i);
    
    if(strptime(buf,"%Y%m%d%H",&tm) == NULL)
        return RC_FAIL;

    thistime = mktime(&tm);
    if(thistime < 0 )
        return RC_FAIL;

#if 0

    if((thistime + MAX_DAYS*24) < baseTime/3600)
    {
        return RC_FAIL;
    }    
#endif
    

    *filetime = thistime;
    return RC_OK;
}

RESULT DATASTOR_validateDBFile(char *fileName)
{
    sqlite3 *db=NULL;
    int rc = 0;
    char nFileName[PATH_MAX];
    int expectLen = (strlen(recorddbpath) + 12);
    int currentlen = strlen(fileName);
    char* pFname = fileName;
     
    // recorddbpath /20110101/23    
    if( expectLen < currentlen) 
    {
        memcpy(nFileName, fileName, expectLen);
        nFileName[expectLen] = '\0';
        pFname = nFileName;
       
    }
    else if(expectLen > currentlen)
    {
        return RC_FAIL;
    }
    
    //the file we needed
    rc = sqlite3_open(pFname,&db);
    if(rc )
    {
        DATAMLogError("Can't open database: %s", sqlite3_errmsg(db));
        sqlite3_close(db);
        return RC_FAIL;
    }
    
    //rc = sqlite3_key(db,dbstorekey,6);
    if(SQLITE_OK != rc)
    {
        DATAMLogError("Can't open database: %s", sqlite3_errmsg(db));
        sqlite3_close(db);
        
        return RC_FAIL;
    }

    rc = sqlite3_exec(db,sql_creatdb,0,0,NULL);
    if(rc)
    {
        sqlite3_close(db);
        return RC_FAIL;
    }
    
    sqlite3_close(db);
    return RC_OK;
}

RESULT DATASTOR_analyzeDBTime(U64 dbtime, U64* mintime, U64*maxtime)
{
    struct tm thistm ; 
    char filename[PATH_MAX];
    int rc = 0;
    sqlite3* db = NULL;
    char **dbResult;
    int nRow;
    int nColumn;
    char sql[256];

    if(dbtime == 0)
        return RC_FAIL;
    
    localtime_r((time_t*)(&dbtime),&thistm);

    sprintf(filename, "%s/%04d%02d%02d/%02d", recorddbpath,1900+thistm.tm_year,1+thistm.tm_mon,thistm.tm_mday,thistm.tm_hour);
    //printf("open db : %s \n",filename);
    rc =sqlite3_open(filename,&db );
    if(rc )
    {
        DATAMLogError("Can't open database: %s", sqlite3_errmsg(db));
        sqlite3_close(db);     
        return RC_FAIL;
    }

 //   rc = sqlite3_key(db,dbstorekey,6);
    if(SQLITE_OK != rc)
    {
        DATAMLogError("Can't open database: %s", sqlite3_errmsg(db));
        sqlite3_close(db);      
        return RC_FAIL;
    }

    sprintf(sql,"select MIN(TIMESTAMP),MAX(TIMESTAMP) from PHONE_INFO;");

    rc = sqlite3_get_table(db,sql,&dbResult,&nRow,&nColumn,NULL);
    if(rc)
    {
       DATAMLogError("Get record info failed %s", sqlite3_errmsg(db));
       sqlite3_free_table(dbResult);
       sqlite3_close(db);   
       return RC_FAIL;
    }

    // empty db is an abnormal case. it should not happen.
    if(dbResult[2] && dbResult[3])
    {
        if(mintime)  *mintime= strtoull(dbResult[2],NULL,10);
        if(maxtime) *maxtime= strtoull(dbResult[3],NULL,10);
    }
    else
    {
        DATAMLogError("Empty database!",NULL );
        sqlite3_free_table(dbResult);
        sqlite3_close(db);  
        return RC_FAIL;   
    }
    
     sqlite3_free_table(dbResult);
     sqlite3_close(db);  
     return RC_OK;   
}


RESULT DATASTOR_openStorage(void)
{
    char subpath[PATH_MAX];
    char filename[PATH_MAX];
    DIR *pDir, *pDayDir;
    struct dirent *dt = NULL;
    struct dirent *subdt = NULL;
    time_t filetime;
    pDir = opendir(recorddbpath);
    if(pDir == NULL)
    {
        if (mkdir(recorddbpath,0777) < 0)
        {
            DATAMLogError("creat dir %s failed!!!",recorddbpath);
            return RC_FAIL;
        }
        else
        {
            pDir = opendir(recorddbpath);
        }       
    }

    while((dt = readdir(pDir)) != NULL)
    {
        if((strcmp(dt->d_name,".") == 0) || (strcmp(dt->d_name,"..") == 0))
            continue;

        sprintf(subpath,"%s/%s", recorddbpath,dt->d_name);
        if( (dt->d_type != DT_DIR)  
            || (DATASTOR_validateDirName(dt->d_name) == RC_FAIL))
        {
            char cmd[PATH_MAX + 32] ; 
            sprintf(cmd,"rm -rf %s", subpath);
            system(cmd);   
            continue;
        }

        //right
        pDayDir = opendir(subpath);
        while((subdt = readdir(pDayDir)) != NULL)
        {
            if((strcmp(subdt->d_name,".") == 0) || (strcmp(subdt->d_name,"..") == 0))
                continue;

            sprintf(filename,"%s/%s", subpath,subdt->d_name);
            if( (subdt->d_type != DT_REG)  
                || (DATASTOR_validateFileName(dt->d_name,subdt->d_name,&filetime) == RC_FAIL)
                || (DATASTOR_validateDBFile(filename) == RC_FAIL))
            {
                char cmd[PATH_MAX + 32] ; 
                sprintf(cmd,"rm -rf %s", filename);
                system(cmd);   
                continue;
            }

            DATASTOR_updateTimeRange(filetime);
        }
    
        closedir(pDayDir);
        
    }
    closedir(pDir);

    return RC_OK;
}

void  DATASTOR_updateDBDesc()
{
 
    if(recorddb_desc.range_begin && recorddb_desc.range_end)
    {
         DATASTOR_analyzeDBTime(recorddb_desc.range_end, NULL,&recorddb_desc.range_end);
         DATASTOR_analyzeDBTime(recorddb_desc.range_begin, &recorddb_desc.range_begin, NULL);
    }
}

RESULT DATASTOR_init(void)
{
	pthread_t pthrd;
    MODREG modReg = {0};

    sprintf(recorddbpath, "%s/recorddb",data_storage_path);
    
    needPostProc = FALSE;
    
    DATASTOR_initFlowCtrl();

    recorddb_desc.range_begin = recorddb_desc.range_end = 0;
    if(DATASTOR_openStorage() != RC_OK)
        return RC_FAIL;

    DATASTOR_updateDBDesc();
    DATAMLogNotice("storage read completed: begin = %llu, end = %llu ", recorddb_desc.range_begin, recorddb_desc.range_end);
    
    modReg.modID = MODULE_DATAM;
    modReg.needTick = TRUE;
    modReg.tickPeriod = DATASTOR_TIMER_RESOLUTION; // 10s
    modReg.needModTimer = TRUE;

    MOD_register(&modReg);
       
    pthread_create(&pthrd, NULL, DATASTOR_main, NULL);

    return RC_OK;


}

void DATASTOR_printStats(void)
{
    int slotNum = 0;
    static U32 last_UEIDSaved = 0;
    static U32 last_UEIDGotten = 0;
    static U32 last_slotSaved[10] = {0};
    U32 diff_saved = datamStats.numUeIDSaved - last_UEIDSaved;
    U32 diff_gotten = datamStats.numUeIDGotten - last_UEIDGotten;
    last_UEIDSaved = datamStats.numUeIDSaved; 
    last_UEIDGotten = datamStats.numUeIDGotten;
    dbgprintf_s(DEBUG_INFO, stdout, "********** MODID[0x%x] Data Storage statistics (dec) *********************\n ", MODULE_DATAM);
    dbgprintf_s(DEBUG_INFO, stdout, "UEIDGotten = %u, UEIDSaved = %u, UEIDGotten_Inc = %u, UEIDSaved_Inc = %u\n",datamStats.numUeIDGotten,datamStats.numUeIDSaved, diff_gotten,diff_saved);
    for (slotNum = 0; slotNum < 10 ; slotNum++)
    {
        dbgprintf_s(DEBUG_INFO, stdout, "BBU[%d] UEIDGotten = %u UEIDGotten_Inc = %u \n", (slotNum + 1), datamStats.numBbuSaved[slotNum], (datamStats.numBbuSaved[slotNum] - last_slotSaved[slotNum]));
        last_slotSaved[slotNum] = datamStats.numBbuSaved[slotNum];
    }

}


